package com.zhss.dfs.namenode.server;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 负责管理datanode组件
 */
public class DataNodeManager {


    private ConcurrentHashMap<String, DataNodeInfo> dataNodes = new ConcurrentHashMap<>();

    private List<String> toRemoveDatanodes = new ArrayList<>();

    public DataNodeManager() {
        new DataNodeAliveMonitor().start();
    }

    /**
     * dataNode进行注册
     *
     * @param ip
     * @param hostName
     */
    public Boolean register(String ip, String hostName) {
        DataNodeInfo dataNodeInfo = new DataNodeInfo(ip, hostName);
        dataNodes.put(ip + "-" + hostName, dataNodeInfo);
        return true;
    }


    class DataNodeAliveMonitor extends Thread {
        @Override
        public void run() {
            try {
                while (true) {
                    Iterator<DataNodeInfo> iterator = dataNodes.values().iterator();
                    DataNodeInfo dataNodeInfo = null;
                    while (iterator.hasNext()) {
                        DataNodeInfo nodeInfo = iterator.next();
                        if (System.currentTimeMillis() - dataNodeInfo.getLastestHeartbeatTime() > 90 * 1000) {
                            toRemoveDatanodes.add(dataNodeInfo.getIp() + "-" + dataNodeInfo.getHostName());
                        }
                    }
                    if (!toRemoveDatanodes.isEmpty()) {
                        for (String toRemove : toRemoveDatanodes) {
                            dataNodes.remove(toRemove);
                        }
                    }
                    Thread.sleep(30 * 1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


}
